package com.zt.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zt.mapper.GoodsMapper;
import com.zt.pojo.Goods;
import com.zt.service.GoodsService;
import com.zt.utils.TextModerationPlusUtil;
import com.zt.utils.common.dtos.ResponseResult;
import com.zt.utils.common.enums.AppHttpCodeEnum;
import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.Set;

/**
* @author dell
* @description 针对表【goods】的数据库操作Service实现
* @createDate 2024-09-04 10:51:24
*/
@Service
public class GoodsServiceImpl extends ServiceImpl<GoodsMapper, Goods>
    implements GoodsService{
    @Autowired
    private GoodsMapper goodsMapper;

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    @Resource
    private RedisTemplate<String,Object> redisTemplate=new RedisTemplate<>();

    @SneakyThrows
    @Override
    public ResponseResult saveAndAuto(Goods goods) {
        if (ObjectUtils.isEmpty(goods)) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
//        1.对内容进行审核
        ResponseResult content = TextModerationPlusUtil.content(goods.getName());

        if(content.getCode()==600){
//            1.1审核通过,修改状态
            goods.setApSatus(1);
            goods.setPutStatus(1);
//            1.2 添加到数据库中
            goodsMapper.insert(goods);

            /*2.
             * 根据发布时间进行与当前时间作比对,
             * 如果小于等于当前时间，存放到kafka,
             * 如果大于当前时间并小于五分钟放到redis中
             */
//            发布时间
            long putTime = goods.getPutTime().getTime();
//            当前时间
            long CurrentTime = System.currentTimeMillis();
//            五分钟后的时间
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE,5);
            long FiveMinutes = calendar.getTimeInMillis();

            if(putTime<=CurrentTime){
//                放到kafka中 Instantly--立刻的意思
                kafkaTemplate.send("Instantly", JSON.toJSONString(goods));
            } else if (putTime <= FiveMinutes) {
//                放到redis中
                redisTemplate.opsForZSet().add("Five:goods:",goods,goods.getPutTime().getTime());
            }

        } else if (content.getCode()==601) {
            goods.setPutStatus(0);
            goodsMapper.insert(goods);
            return ResponseResult.errorResult(5001,"转人工处理");
        }else{
            return ResponseResult.errorResult(501,"未通过审核");
        }
        return ResponseResult.okResult("操作成功");
    }


//    每一分钟扫描一次redis集合
    @Scheduled(cron = "0 0/1 * * * ?")
    public void RedisToKafka(){
//        获取当前时间
        long CurrentTime = System.currentTimeMillis();
        Set<Object> set = redisTemplate.opsForZSet().rangeByScore("Five:goods:", 0, CurrentTime);
//        1.判断集合是否为空
        if(set.size()!=0&&set!=null){
//            遍历集合
            for (Object o : set) {
//                存入到kafka中
                kafkaTemplate.send("Instantly",JSON.toJSONString(o));
//                删除set中的数据
                redisTemplate.opsForZSet().remove("Five:goods:",o);
            }
        }
    }

}




